#include "market_data/market_data_types.h"
#include <spdlog/spdlog.h>
#include <sstream>
#include <iomanip>
#include <filesystem>
#include <fstream>
#include <ctime>
#include <memory>
#include <queue>
#include <condition_variable>

namespace hft {

// 使用智能指针管理QuestDB连接
class QuestDBConnection {
public:
    QuestDBConnection(const std::string& host, int port) {
        line_tcp_ = questdb_ilp_line_tcp_create();
        if (!line_tcp_) {
            throw std::runtime_error("Failed to create QuestDB line TCP client");
        }
        if (!questdb_ilp_line_tcp_connect(line_tcp_, host.c_str(), port)) {
            questdb_ilp_line_tcp_close(line_tcp_);
            throw std::runtime_error("Failed to connect to QuestDB server");
        }
    }

    ~QuestDBConnection() {
        if (line_tcp_) {
            questdb_ilp_line_tcp_close(line_tcp_);
        }
    }

    questdb_ilp_line_tcp_t* get() { return line_tcp_; }

private:
    questdb_ilp_line_tcp_t* line_tcp_;
};

// 数据写入缓冲�?
class DataBuffer {
public:
    DataBuffer(size_t max_size = 10000) : max_size_(max_size) {}

    void add(const MarketDataMessage& msg) {
        std::lock_guard<std::mutex> lock(mutex_);
        buffer_.push(msg);
        if (buffer_.size() >= max_size_) {
            cv_.notify_one();
        }
    }

    bool get(MarketDataMessage& msg) {
        std::lock_guard<std::mutex> lock(mutex_);
        if (buffer_.empty()) {
            return false;
        }
        msg = buffer_.front();
        buffer_.pop();
        return true;
    }

    void wait_for_data() {
        std::unique_lock<std::mutex> lock(mutex_);
        cv_.wait(lock, [this] { return !buffer_.empty(); });
    }

    size_t size() const {
        std::lock_guard<std::mutex> lock(mutex_);
        return buffer_.size();
    }

private:
    std::queue<MarketDataMessage> buffer_;
    mutable std::mutex mutex_;
    std::condition_variable cv_;
    const size_t max_size_;
};

// 单例实现
DataManager& DataManager::getInstance() {
    static DataManager instance;
    return instance;
}

bool DataManager::initialize(const std::string& host, int port) {
    std::lock_guard<std::mutex> lock(mutex_);
    
    try {
        // 创建QuestDB连接
        connection_ = std::make_unique<QuestDBConnection>(host, port);
        
        // 初始化时间戳
        last_compression_time_ = std::chrono::system_clock::now();
        last_backup_time_ = std::chrono::system_clock::now();

        // 启动写入线程
        write_thread_ = std::thread(&DataManager::writeThread, this);
        
        spdlog::info("Connected to QuestDB server at {}:{}", host, port);
        return true;
    } catch (const std::exception& e) {
        spdlog::error("Failed to initialize QuestDB connection: {}", e.what());
        cleanup();
        return false;
    }
}

void DataManager::onTick(const MarketDataMessage& msg) {
    // 将数据添加到缓冲�?
    data_buffer_.add(msg);
    
    // 更新最新数�?
    std::lock_guard<std::mutex> lock(mutex_);
    latest_data_[msg.symbol] = msg;

    // 检查是否需要压缩数�?
    auto now = std::chrono::system_clock::now();
    if (compression_config_.enabled && 
        (now - last_compression_time_) >= compression_config_.compression_interval) {
        compressData();
        last_compression_time_ = now;
    }

    // 检查是否需要备份数�?
    if (backup_config_.enabled && 
        (now - last_backup_time_) >= backup_config_.backup_interval) {
        backupData();
        last_backup_time_ = now;
    }
}

void DataManager::writeThread() {
    while (!stop_write_thread_) {
        try {
            // 等待数据
            data_buffer_.wait_for_data();
            
            // 批量写入数据
            std::vector<MarketDataMessage> batch;
            MarketDataMessage msg;
            while (data_buffer_.get(msg) && batch.size() < 1000) {
                batch.push_back(msg);
            }
            
            if (!batch.empty()) {
                writeBatch(batch);
            }
        } catch (const std::exception& e) {
            spdlog::error("Error in write thread: {}", e.what());
            std::this_thread::sleep_for(std::chrono::seconds(1));
        }
    }
}

void DataManager::writeBatch(const std::vector<MarketDataMessage>& batch) {
    std::lock_guard<std::mutex> lock(mutex_);
    
    if (!connection_) {
        throw std::runtime_error("QuestDB connection not initialized");
    }

    for (const auto& msg : batch) {
        try {
            // 将时间戳转换为纳�?
            auto timestamp_ns = std::chrono::duration_cast<std::chrono::nanoseconds>(
                msg.timestamp.time_since_epoch()).count();

            // 创建QuestDB�?
            questdb_ilp_line_tcp_at(connection_->get(), timestamp_ns);
            questdb_ilp_line_tcp_symbol(connection_->get(), "symbol", msg.symbol.c_str());
            questdb_ilp_line_tcp_at_field(connection_->get(), "price", msg.price);
            questdb_ilp_line_tcp_at_field(connection_->get(), "volume", msg.volume);
            questdb_ilp_line_tcp_at_field(connection_->get(), "turnover", msg.turnover);
            questdb_ilp_line_tcp_at_field(connection_->get(), "open_interest", msg.open_interest);
            questdb_ilp_line_tcp_at_field(connection_->get(), "bid_price", msg.bid_price);
            questdb_ilp_line_tcp_at_field(connection_->get(), "bid_volume", msg.bid_volume);
            questdb_ilp_line_tcp_at_field(connection_->get(), "ask_price", msg.ask_price);
            questdb_ilp_line_tcp_at_field(connection_->get(), "ask_volume", msg.ask_volume);
        } catch (const std::exception& e) {
            spdlog::error("Failed to write market data to QuestDB: {}", e.what());
        }
    }

    // 批量刷新数据
    if (!questdb_ilp_line_tcp_flush(connection_->get())) {
        throw std::runtime_error("Failed to flush data to QuestDB");
    }
}

void DataManager::cleanup() {
    std::lock_guard<std::mutex> lock(mutex_);
    
    // 停止写入线程
    stop_write_thread_ = true;
    if (write_thread_.joinable()) {
        write_thread_.join();
    }
    
    // 停止回放
    stopReplay();
    
    // 清理连接
    connection_.reset();
}

QueryResult DataManager::queryData(const std::string& symbol,
                                 const std::chrono::system_clock::time_point& start_time,
                                 const std::chrono::system_clock::time_point& end_time) {
    QueryResult result;
    
    try {
        // 构建查询语句
        std::stringstream ss;
        ss << "SELECT * FROM market_data WHERE symbol = '" << symbol << "' "
           << "AND timestamp BETWEEN " 
           << std::chrono::duration_cast<std::chrono::nanoseconds>(
                  start_time.time_since_epoch()).count() << " AND "
           << std::chrono::duration_cast<std::chrono::nanoseconds>(
                  end_time.time_since_epoch()).count()
           << " ORDER BY timestamp ASC";
        
        executeQuery(ss.str(), result);
    } catch (const std::exception& e) {
        result.error = e.what();
        spdlog::error("Failed to query data: {}", e.what());
    }
    
    return result;
}

void DataManager::startReplay(const ReplayConfig& config,
                            std::function<void(const MarketDataMessage&)> callback) {
    if (replay_running_) {
        stopReplay();
    }
    
    replay_running_ = true;
    replay_thread_ = std::thread([this, config, callback]() {
        try {
            // 查询数据
            auto result = queryData(config.symbol, config.start_time, config.end_time);
            if (!result.error.empty()) {
                throw std::runtime_error(result.error);
            }
            
            // 回放数据
            for (const auto& msg : result.data) {
                if (!replay_running_) {
                    break;
                }
                
                callback(msg);
                
                if (!config.real_time) {
                    std::this_thread::sleep_for(config.interval);
                }
            }
        } catch (const std::exception& e) {
            spdlog::error("Replay failed: {}", e.what());
        }
        
        replay_running_ = false;
    });
}

void DataManager::stopReplay() {
    replay_running_ = false;
    if (replay_thread_.joinable()) {
        replay_thread_.join();
    }
}

void DataManager::setCompressionConfig(const CompressionConfig& config) {
    std::lock_guard<std::mutex> lock(mutex_);
    compression_config_ = config;
}

void DataManager::compressData() {
    std::lock_guard<std::mutex> lock(mutex_);
    
    try {
        // 计算压缩时间范围
        auto now = std::chrono::system_clock::now();
        auto start_time = now - compression_config_.retention_period;
        
        // 构建压缩查询
        std::stringstream ss;
        ss << "COMPRESS TABLE market_data "
           << "WHERE timestamp < " 
           << std::chrono::duration_cast<std::chrono::nanoseconds>(
                  start_time.time_since_epoch()).count()
           << " WITH RATIO " << compression_config_.compression_ratio;
        
        QueryResult result;
        executeQuery(ss.str(), result);
        
        if (!result.error.empty()) {
            throw std::runtime_error(result.error);
        }
        
        spdlog::info("Data compression completed successfully");
    } catch (const std::exception& e) {
        spdlog::error("Data compression failed: {}", e.what());
    }
}

void DataManager::setBackupConfig(const BackupConfig& config) {
    std::lock_guard<std::mutex> lock(mutex_);
    backup_config_ = config;
}

void DataManager::backupData() {
    std::lock_guard<std::mutex> lock(mutex_);
    
    try {
        // 创建备份目录
        std::filesystem::create_directories(backup_config_.backup_dir);
        
        // 生成备份文件�?
        std::string backup_file = generateBackupFilename();
        std::string backup_path = backup_config_.backup_dir + "/" + backup_file;
        
        // 导出数据
        std::stringstream ss;
        ss << "EXPORT TABLE market_data TO '" << backup_path << "'";
        
        QueryResult result;
        executeQuery(ss.str(), result);
        
        if (!result.error.empty()) {
            throw std::runtime_error(result.error);
        }
        
        // 清理旧备�?
        cleanupOldBackups();
        
        spdlog::info("Data backup completed successfully: {}", backup_path);
    } catch (const std::exception& e) {
        spdlog::error("Data backup failed: {}", e.what());
    }
}

void DataManager::cleanupOldBackups() {
    try {
        std::vector<std::filesystem::path> backups;
        for (const auto& entry : std::filesystem::directory_iterator(backup_config_.backup_dir)) {
            if (entry.is_regular_file()) {
                backups.push_back(entry.path());
            }
        }
        
        // 按修改时间排�?
        std::sort(backups.begin(), backups.end(),
                 [](const auto& a, const auto& b) {
                     return std::filesystem::last_write_time(a) > std::filesystem::last_write_time(b);
                 });
        
        // 删除多余的备�?
        while (backups.size() > backup_config_.max_backups) {
            std::filesystem::remove(backups.back());
            backups.pop_back();
        }
    } catch (const std::exception& e) {
        spdlog::error("Failed to cleanup old backups: {}", e.what());
    }
}

std::string DataManager::generateBackupFilename() const {
    auto now = std::chrono::system_clock::now();
    auto time = std::chrono::system_clock::to_time_t(now);
    std::stringstream ss;
    ss << "backup_" << std::put_time(std::localtime(&time), "%Y%m%d_%H%M%S") << ".csv";
    return ss.str();
}

// 私有辅助函数实现
void DataManager::executeQuery(const std::string& query, QueryResult& result) {
    // 这里需要实现QuestDB的查询执�?
    // 由于QuestDB的C++客户端API限制，这里需要调用QuestDB的REST API
    // 实际实现时需要使用HTTP客户端库（如libcurl）来执行查询
    // 这里仅作为示�?
    spdlog::debug("Executing query: {}", query);
}

} // namespace hft 